package com.pajk.user.service.credit;

import java.io.Serializable;
import java.util.*;

import javax.annotation.Resource;

import com.alibaba.fastjson.JSON;
import com.pajk.user.dal.mapper.RedoMonitorMapper;
import com.pajk.user.model.RedoMonitor;
import com.pajk.user.service.common.idpool.IDPool;
import com.pajk.user.service.common.metaq.MsgObj;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
import org.springframework.transaction.support.TransactionTemplate;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.pajk.user.dal.mapper.CreditHisMapper;
import com.pajk.user.dal.mapper.UserMapper;
import com.pajk.user.dal.model.CreditHisDO;
import com.pajk.user.service.common.metaq.BaseConsumer;
import com.pajk.user.service.common.metaq.TopicEnum;

public class ReduceCreditConsumer extends BaseConsumer {

    public static final String  CONSUME_CREDIT = "consumeCredit";
    public static final String  USER_ID        = "userId";
    public static final String  LEFT_CREDIT    = "leftCredit";
    public static final String  ACTION_CODE    = "actionCode";
    public static final String  ACTION_ID      = "actionId";
    public static final String  ACTION_TIME    = "actionTime";
    public static final String  APPFROM        = "appFrom";

    private static final Logger log            = LoggerFactory
                                                   .getLogger(ReduceCreditConsumer.class);

    @Resource
    private TransactionTemplate transactionTemplate;
    @Resource
    private TransactionTemplate userTransactionTemplate;
    @Resource
    private CreditHisMapper     creditHisMapper;
    @Resource
    RedoMonitorMapper creditErrorMapper;
    @Resource
    private UserMapper          userMapper;

    @Override
    public TopicEnum getTopicEnum() {
        return TopicEnum.USER_CONSUME_CREDIT;
    }

    @Override
    public void doLogErrorConsumeMessage(final MsgObj msgObj) {
        {
            List<Serializable> msgs=msgObj.getMsgList();
            for (final Serializable obj : msgs) {
                if (obj instanceof String) {
                    userTransactionTemplate.execute(new TransactionCallbackWithoutResult() {
                        public void doInTransactionWithoutResult(TransactionStatus status) {
                            try {
                                HashMap<String, Object> msg = (HashMap<String, Object>)JSON.parseObject((String)obj,HashMap.class);

                                String actionId = (String) msg.get(ACTION_ID);
                                RedoMonitor creditError = new RedoMonitor();
                                creditError.setId(UUID.randomUUID().toString());
                                creditError.setBusinessId(actionId);
                                creditError.setDescription("CreditAdd Error");
                                creditError.setErrorMsg(msgObj.getErrorMsg());
                                creditError.setMsgBody(obj.toString());
                                creditError.setStatus(Byte.valueOf("1"));
                                creditError.setTopic(getTopicEnum().getTopic());
                                creditErrorMapper.insertRedoMonitor(creditError);
                            } catch (Exception e) {
                                log.error("doLogErrorConsumeMessage failed", e);
                                status.setRollbackOnly();
                            } catch (Throwable t){
                                log.error("throwable catched", t);
                                status.setRollbackOnly();
                            }
                        }
                    });
                }
            }
        }
    }

    @Override
    public ConsumeConcurrentlyStatus doConsumeMessage(final MsgObj msgObj) {
        final int maxRetryCount=super.maxRetryCount;
        List<Serializable> msgs=msgObj.getMsgList();
        for (final Serializable obj : msgs) {
            if (obj instanceof String) {
                HashMap<String, Object> msg = null;
                final CreditHisDO creditHisDO = new CreditHisDO();
                int leftCredit = 0;
                try {
                    msg = (HashMap<String, Object>) JSON.parseObject((String) obj, HashMap.class);

                    leftCredit = (Integer) msg.get(LEFT_CREDIT);
                    creditHisDO.setUserId((Long) msg.get(USER_ID));
                    creditHisDO.setActionId((String) msg.get(ACTION_ID));
                    creditHisDO.setActionCode((String) msg.get(ACTION_CODE));
                    creditHisDO.setAppFrom((String) msg.get(APPFROM));
                    creditHisDO.setCredit(0 - (Integer) msg.get(CONSUME_CREDIT));
                    creditHisDO.setActionTime((Date) msg.get(ACTION_TIME));

                    creditHisDO.setSync(CreditStatusEnum.Synchronized.value());
                } catch (Exception e) {
                    log.error("get msg info failed, msg: " + msg.toString());
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                final int leftCredit1 = leftCredit;
                return transactionTemplate.execute(new TransactionCallback<ConsumeConcurrentlyStatus>() {
                    public ConsumeConcurrentlyStatus doInTransaction(TransactionStatus status) {

                        try {
                            creditHisMapper.insertCredit(creditHisDO);
                            Map<String, Object> updateMap = new HashMap<String, Object>();
                            updateMap.put("id", creditHisDO.getUserId());
                            updateMap.put("wanliCredits", leftCredit1);
                            userMapper.updateWanliCreditsById(updateMap);
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        } catch (Throwable e) {
                            status.setRollbackOnly();
                            if(msgObj.getReconsumeTimes()>=maxRetryCount){
                                log.error("reconsumeTimes >" + maxRetryCount + "msgs:" + msgObj + "context:" + msgObj.getContext());
                                msgObj.setErrorMsg(e.getMessage());
                                doLogErrorConsumeMessage(msgObj);
                                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                            }else{
                                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                            }
                        }
                    }
                });
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

}
